package org.lzfto.flink.demo.config.mqtt;

import java.util.UUID;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.lzfto.flink.demo.config.AppConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;

/**
 * MQTT配置
 * </br>生产环境，不推荐使用管理员连接，除非有特殊目的
 * @author lzfto
 * @created 2025年6月13日 11:41:23
 */
public class MQTTConfig {

	private static final  Logger LOGGER=LoggerFactory.getLogger(MQTTConfig.class);

	
	private final String CONSUMER_CHANNEL_BEAN_NAME="mqttConsumerboundChannel";
	private final String CONSUMER_ADAPTER_BEAN_NAME="mqttConsumerboundAdapter";
	private final String CONSUMER_DSL_BEAN_NAME="mqttConsumerDslFlow";
	
	public final static  String PRODUCER_CHANNEL_BEAN_NAME="mqttProducerboundChannel";
	private final String PRODUCER_HANDLER_BEAN_NAME="mqttProducerboundHandler";
	private final String PRODUCER_DSL_BEAN_NAME="mqttProducerDslFlow";
	
	private final String MESSAGE_TEMPLATE_BEAN_NAME="artemisMessagingTemplate";
	
	
    private static MqttPahoMessageDrivenChannelAdapter messageDrivenAdapter=null;
	private DefaultListableBeanFactory beanFactory;
	private AppConfiguration config;

	public MQTTConfig(DefaultListableBeanFactory beanFactory, AppConfiguration config) {
		this.beanFactory = beanFactory;
		this.config = config;
	}
	
	public static void stopReceive() {
		if (messageDrivenAdapter!=null) {
			if (messageDrivenAdapter.isActive()) {
				messageDrivenAdapter.stop();
				LOGGER.info("停止接收MQTT消息");
			}
		}
	}
	

	public void init() {
		DefaultMqttPahoClientFactory messageFactory=this.createFactory();
		// 注册发送的
		this.registerProducer(messageFactory);
		// 注册接收的
		// 为了演示MQTT的 Flink source，这里需要暂时断不注册，否则可能会出现问题
		//this.registerConsumer(messageFactory);
		config.setIsArtemisRegistered(true);
	}
	
	private DefaultMqttPahoClientFactory createFactory() {
		DefaultMqttPahoClientFactory messagefactory = new DefaultMqttPahoClientFactory();
		MqttConnectOptions options = new MqttConnectOptions();
		options.setServerURIs(new String[] { config.getArtemisHost() });
		options.setUserName(config.getArtemisUserName());
		options.setPassword(config.getArtemisPassword().toCharArray());
		options.setAutomaticReconnect(config.getAppProp().getMqttAutoReconnect());
		messagefactory.setConnectionOptions(options);
		beanFactory.registerSingleton("mqttPahoClientFactory", messagefactory);
		return messagefactory;
	}

	/**
	 * 注册消费者有关对象
	 * @param messageFactory
	 */
	@SuppressWarnings("unused")
	private void registerConsumer(DefaultMqttPahoClientFactory messageFactory) {
		// 1.0 创建输入通道
		DirectChannel inboundChannel = new DirectChannel();
		beanFactory.registerSingleton(CONSUMER_CHANNEL_BEAN_NAME, inboundChannel);
		inboundChannel.subscribe(new CustMqttInMessageHandler());
		
		// 2.0 创建消息驱动适配器（消费者客户端）
		messageDrivenAdapter = new MqttPahoMessageDrivenChannelAdapter(
				this.config.getAppProp().getMqttConsumerClientName(),// 唯一客户端ID
				messageFactory,
				this.config.getAppProp().getMqttTopicPrefix() // 订阅主题
		);
		messageDrivenAdapter.setCompletionTimeout(5000);
		messageDrivenAdapter.setConverter(new DefaultPahoMessageConverter());
		messageDrivenAdapter.setQos(config.getAppProp().getMqttQos()); // 设置消费QoS
		messageDrivenAdapter.setOutputChannel(inboundChannel); // 绑定输入通道
		beanFactory.registerSingleton(CONSUMER_ADAPTER_BEAN_NAME, messageDrivenAdapter);

		// 4.0 关联通道和处理器
		beanFactory.registerSingleton(CONSUMER_DSL_BEAN_NAME, IntegrationFlow
				.from(messageDrivenAdapter) // 从适配器接收消息
				.channel(inboundChannel)
				.get());
		// 5.0 启动适配器（Spring Integration 5.x+ 需要显式启动）
		messageDrivenAdapter.start();
	}

	/**
	 * 注册生产者(消息发送者)
	 * @param messageFactory
	 */
	private void registerProducer(DefaultMqttPahoClientFactory messageFactory) {
		// 1.0 注册mqtt 通道
		DirectChannel channel = new DirectChannel();
		beanFactory.registerSingleton(PRODUCER_CHANNEL_BEAN_NAME, channel);

		// 2.0 注册mqtt 处理器客户端
		String uuid=UUID.randomUUID().toString();
		MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
				this.config.getAppProp().getMqttProducerClientName()+"-"+uuid,
				messageFactory);
		messageHandler.setAsync(true);
		//指定接收者的默认主题，如果不指定就是这个了。
		messageHandler.setDefaultTopic(this.config.getAppProp().getMqttTopicPrefix());
		messageHandler.setDefaultQos(1);
		DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
		messageHandler.setConverter(converter);
		beanFactory.registerSingleton(PRODUCER_HANDLER_BEAN_NAME, messageHandler);

		//3.0 关联通道和处理器
		beanFactory.registerSingleton(PRODUCER_DSL_BEAN_NAME,
				IntegrationFlow.from(PRODUCER_CHANNEL_BEAN_NAME).handle(messageHandler).get());
		channel.subscribe(messageHandler);

		//4.0 注册 MessagingTemplate
		MessagingTemplate messagingTemplate = new MessagingTemplate();
		messagingTemplate.setDefaultChannel(channel); // 绑定到已注册的通道
		beanFactory.registerSingleton(MESSAGE_TEMPLATE_BEAN_NAME, messagingTemplate);
	}

}
